package com.qf.leadnewsarticle.listeners;

import com.qf.leadnewsarticle.service.ArticleService;
import com.qf.qfleadnewsfeignapi.wemedia.WeMediaApi;
import com.qf.qfleadnewsmodel.commons.consts.KafkaConst;
import com.qf.qfleadnewsmodel.commons.consts.RedisConst;
import com.qf.qfleadnewsmodel.wemedia.pojos.WmNews;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class PublishNewsListener {

    @Autowired
    private WeMediaApi weMediaApi;

    @Autowired
    private ArticleService articleService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @KafkaListener(topics = KafkaConst.NEWS_PUBLISH_TOPIC)
    public void pubNews(String newsId, Acknowledgment ack){
        log.info("----------收到发布文章消息，消息id为:{}",newsId);

        //根据当前收到的newsId查询wmnews信息
        WmNews wmNews = weMediaApi.findWmNewsById(Integer.parseInt(newsId));
        if (wmNews != null){
            log.info("----------查询到的wmnews信息：{}",wmNews);

            String key = RedisConst.NEWS_PUBLISH_MD_PREFIX + newsId;

            Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, "");
            redisTemplate.expire(key,2, TimeUnit.MINUTES);

            if (flag) {
                try {
                    articleService.publishNews(wmNews);
                } catch (Exception e) {
                    //todo: 记录失败重试的次数，达到3次，就将这个消息记录到消息表中。后期利用定时任务进行消费
                    //删除key
                    redisTemplate.delete(key);
                    //不确认
                    ack.nack(0);
                }
            }
        }
        //手动确认
        ack.acknowledge();
    }
}
